package com.webank.wedatasphere.qualitis.handler;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.httpclient.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.HashMap;

@Service
public class CommonExecuteTaskService {

    private Logger logger = LoggerFactory.getLogger(CommonExecuteTaskService.class);


    // 注入RestTemplate实例
    @Autowired
    private RestTemplate restClient;


    @Value("${linkis.datasource.url:http://10.130.1.37:8188}")
    private String linkisUrl;


    @Value("${linkis.datasource.username:hadoop}")
    private String username;

    @Value("${linkis.datasource.password:f45b9a1af}")
    private String password;

    /**
     * 登录远程的Linkis服务
     *
     * @return ResponseEntity
     */
    public ResponseEntity<JSONObject> login() {
        JSONObject postData = new JSONObject();
        postData.put("userName", username);
        postData.put("password", password);
        String loginUrl = linkisUrl + "/api/rest_j/v1/user/login";
        return restClient.postForEntity(loginUrl, postData, JSONObject.class);
    }


    /**
     * 执行任务
     *
     * @param sql 要执行的sql代码
     * @return ResponseEntity
     */
    public ResponseEntity<JSONObject> executeSql(String sql) {
        String url = "/api/rest_j/v1/entrance/execute";
        JSONObject map = new JSONObject();
        map.put("method", url);
        //用户指定的运行服务程序的参数,必填，里面的值可以为空
        map.put("params", new HashMap<>());
        //执行引擎，用的spark
        map.put("executeApplicationName", "spark");
        map.put("executionCode", sql);
        //当用户执行如spark服务时，可以选择python、R、SQL等,不能为空
        map.put("runType", "sql");
        //因为我没有执行文件脚本，所以没有scriptPath参数
        String executeSql = linkisUrl + url;
        return restClient.postForEntity(executeSql, map, JSONObject.class);
    }

    /**
     * 获取执行任务的状态
     *
     * @param execID execID
     * @return ResponseEntity
     * @throws Exception Exception
     */
    public String getTaskStatus(String execID) {
        String statusUrl = linkisUrl + "/api/rest_j/v1/entrance/" + execID + "/status";
        ResponseEntity<JSONObject> statusResp = restClient.getForEntity(statusUrl, JSONObject.class);
        if (statusResp != null && statusResp.getStatusCode().value() == HttpStatus.SC_OK) {
            String status = statusResp.getBody().getJSONObject("data").getString("status");
            if ("Succeed".equals(status)) {
                return "Succeed";
            }
            if ("Failed".equals(status)) {
                return "Failed";
            }
        }
        return "Running";
    }

    /**
     * 获取执行任务的状态|需要强制执行完成之后才返回结果：不管是成功还是失败
     *
     * @param execID execID
     * @return ResponseEntity
     * @throws Exception Exception
     */
    public String getTaskStatusWaitTime(String execID, long thisTime) throws InterruptedException {
        String statusUrl = linkisUrl + "/api/rest_j/v1/entrance/" + execID + "/status";
        ResponseEntity<JSONObject> statusResp = restClient.getForEntity(statusUrl, JSONObject.class);
        if (statusResp != null && statusResp.getStatusCode().value() == HttpStatus.SC_OK) {
            String status;
            for (; ; ) {
                // 持续刷新状态
                statusResp = restClient.getForEntity(statusUrl, JSONObject.class);
                status = statusResp.getBody().getJSONObject("data").getString("status");
                //死循环查看任务状态，如果任务成功或者失败，则退出循环
                if ("Succeed".equals(status) || "Failed".equals(status)) {
                    break;
                }
                // 睡眠6秒钟
                Thread.sleep(6000);
                // 自选最大阈值十分钟
                if (thisTime <= (System.currentTimeMillis() - (10 * 60 * 1000))) {
                    logger.error("超时十分钟还未执行完成。。。。。。。自动退出。。。。");
                    break;
                }
            }
            if ("Succeed".equals(status)) {
                // do something
                return "Succeed";
            }else {
                return "Failed";
            }
        }
        return "Running";
    }


    /**
     * 获取执行任务的的结果文件位置
     *
     * @param taskID taskID
     * @return String
     */
    public JSONObject getFIleURL(String taskID) {
        String historyUrl = "/api/rest_j/v1/jobhistory/" + taskID + "/get";
        String executeSql = linkisUrl + historyUrl;
        ResponseEntity<JSONObject> hisResp = restClient.getForEntity(executeSql, JSONObject.class);
        JSONObject resultLocation = null;
        if (hisResp != null && hisResp.getStatusCode().value() == HttpStatus.SC_OK) {
            resultLocation = hisResp.getBody().getJSONObject("data").getJSONObject("task");
        }
        return resultLocation;
    }


    /**
     * 获取结果
     *
     * @param resultLocation
     * @return
     * @throws Exception
     */
    public JSONArray getResult(String resultLocation) {
        // 下载保存路径·下文件的类型
        String resUrl = linkisUrl + "/api/rest_j/v1/filesystem/openFile?path=" + resultLocation + "/_0.dolphin";
        ResponseEntity<JSONObject> resResp = restClient.getForEntity(resUrl, JSONObject.class);
        if (resResp != null && resResp.getStatusCode().value() == HttpStatus.SC_OK) {
            //do something
            JSONObject body = resResp.getBody();
            assert body != null;
            JSONArray arrayList = body.getJSONObject("data").getJSONArray("fileContent");
            logger.error(arrayList.toJSONString());
            return arrayList;
        }
        return null;
    }

}
